package com.zyl.user;

import java.io.BufferedReader;
import java.io.FileReader;
import java.io.IOException;
import java.util.HashMap;
import org.apache.hadoop.filecache.DistributedCache;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.LongWritable;
import org.apache.hadoop.io.NullWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Mapper;

public class JoinMapper extends Mapper<LongWritable, Text, Text, NullWritable> {
	private HashMap<String, String> user_info = new HashMap();
	private String splitter = "";
	private String rating_secondPart = "";

	public JoinMapper() {
	}

	protected void setup(Mapper<LongWritable, Text, Text, NullWritable>.Context context) throws IOException, InterruptedException {
		this.splitter = context.getConfiguration().get("SPLITTER");
		Path[] distributePaths = DistributedCache.getLocalCacheFiles(context.getConfiguration());
		String line = "";
		BufferedReader br = null;
		Path[] var8 = distributePaths;
		int var7 = distributePaths.length;

		for(int var6 = 0; var6 < var7; ++var6) {
			Path path = var8[var6];
			if (path.toString().endsWith("users.dat")) {
				br = new BufferedReader(new FileReader(path.toString()));

				while((line = br.readLine()) != null) {
					String userID = line.substring(0, line.indexOf("::"));
					String secondPart = line.substring(line.indexOf("::") + 2, line.length());
					this.user_info.put(userID, secondPart);
				}
			}
		}

	}

	protected void map(LongWritable key, Text value, Mapper<LongWritable, Text, Text, NullWritable>.Context context) throws IOException, InterruptedException {
		String[] val = value.toString().split(this.splitter);
		this.rating_secondPart = (String)this.user_info.get(val[0]);
		if (this.rating_secondPart != null) {
			String result = val[0] + this.splitter + this.rating_secondPart + this.splitter + val[1];
			context.write(new Text(result), NullWritable.get());
		}

	}
}
